
package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.BetweenFormater;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrSpliter;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.sparing.dao.SyncRuleConfMapper;
import com.hexinfo.dmpro.sparing.dto.*;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 万国到UAT同步(库级表级)
 *
 * @author yemw
 * @date 2023-09-08 17:45:29
 */
@Service
@AllArgsConstructor
@Slf4j
public class SyncRuleConfServiceFOURImpl extends ServiceImpl<SyncRuleConfMapper, SyncRuleConf> implements SyncRuleConfFOURService {

    private final ScanMetadataService scanMetadataService;

    private final TimeFieldMatchingService timeFieldMatchingService;

    private final HiveToOracleService hiveToOracleService;

    private final RuleToDistCPService ruleToDistCPService;

    private final MessageCenterService messageCenterService;

    private final UatSyncLogsTaskService uatSyncLogsTaskService;

    private final UatSyncLogsService uatSyncLogsService;

    private final AvailabilityPathService availabilityPathService;

    private SyncRuleConfServiceImpl syncRuleConfServiceImpl;

    /**
     * UAT库级表级同步
     * @param handMoveSyncDTO
     * @return
     */
    @Override
    public Map<String,String> uATSyncSearchDatabase(HandMoveSyncDTO handMoveSyncDTO) {
        Map<String,String> map = new HashMap<>();
        //转小写
        handMoveSyncDTO.setDbName(StringUtils.toLowerCase(handMoveSyncDTO.getDbName()));
        handMoveSyncDTO.setTableName(StringUtils.toLowerCase(handMoveSyncDTO.getTableName()));
        String nameNodeStatus = availabilityPathService.getSouAndTar();
        if (StrUtil.isBlank(nameNodeStatus)){
            log.warn("-------------------高可用节点变动异常");
            map.put("error","高可用节点变动异常");
            return map;
        }
//        //消息中心
        MessageCenter mess = new MessageCenter();
        mess.setMenu(ScanCommonConstants.ClusterName.UATRB.value);
        mess.setOperator("admin");
        mess.setDataType(ScanCommonConstants.ClusterName.JXZ.value);
        mess.setMessageWeight(ScanCommonConstants.ClusterName.ONE.value);

        //任务表
        UatSyncLogsTask uatSyncLogsTask = new UatSyncLogsTask();
        uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
        uatSyncLogsTaskService.saveOrUpdate(uatSyncLogsTask);
        //当前开始时间
        Date staDate = DateUtil.date();
        //手动同步库标识
        String prefix = "uat_";
        List<ScanMetadata> scanMetadata = new ArrayList<>();
        //查询该表的全部分区字段
        List<String> hdfsList = new ArrayList<>();
        //获取当前时间分区内的hdfs地址
        List<String> locationList = new ArrayList<>();
        String where = "";
        RuleToDistCPDTO ruleToDistCPDTO = new RuleToDistCPDTO();
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> uatWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");
        ClusterSparingConf uatCluster = scanMetadataService.wrapperClusterSparingConf(uatWrapper,"UAT");
        //DistCP集合
        Set<HdfsHeatBackup> hdfsHeatBackups = new HashSet<>();
        if(StrUtil.isNotBlank(handMoveSyncDTO.getDbName())||StrUtil.isNotBlank(handMoveSyncDTO.getTableName())){
            if(StrUtil.isNotBlank(handMoveSyncDTO.getTableName())){
                mess.setMessageSubject(ScanCommonConstants.ClusterName.CSSJTB.value+"~"+StringUtils.toUpperCase(handMoveSyncDTO.getTableName()));
                messageCenterService.saveOrUpdate(mess);
                //表级同步 查询表信息
                uatSyncLogsTask.setTaskName(StringUtils.toUpperCase(handMoveSyncDTO.getTableName())+DateUtil.format(new Date(), "yyyyMMdd"));
                String[] strings = StrUtil.split(handMoveSyncDTO.getTableName(), StrUtil.DOT);
                String tblDatabaseName = strings[0];
                String tblName = strings[1];
                //全量
                if(StrUtil.isNotBlank(handMoveSyncDTO.getQuantity())&&"全量".equals(handMoveSyncDTO.getQuantity())) {
                    uatSyncLogsTask.setSourceName("【"+handMoveSyncDTO.getLevel()+"】【全量】");
                    //查询该表的hdfs路径
                    List<String> hdfsAlsit = hiveToOracleService.selectHiveLocationUat(tblDatabaseName,tblName);
                    HdfsHeatBackup kup = new HdfsHeatBackup();
                    if(ObjectUtil.isNotEmpty(hdfsAlsit)){
                        String lo = hdfsAlsit.get(0);
                        String  aresoftLo = StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),wGCluster.getHdfsPrefixPath().substring(0,getSlashIndex(wGCluster.getHdfsPrefixPath(),3)));
                        kup.setSourceAddress(aresoftLo);
                        kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),uatCluster.getHdfsPrefixPath().substring(0,getSlashIndex(uatCluster.getHdfsPrefixPath(),3))), "/" + tblDatabaseName , "/" + prefix + tblDatabaseName ));
                        kup.setSourceDbName(tblDatabaseName);
                        kup.setTargetDbName(prefix + tblDatabaseName);
                        kup.setSourceTableName(tblName);
                        kup.setTargetTableName(tblName);
                        hdfsHeatBackups.add(kup);
                    }
                //增量
                }else{
                    uatSyncLogsTask.setSourceName("【"+handMoveSyncDTO.getLevel()+"】【增量】【"+handMoveSyncDTO.getStaDate()+"--"+handMoveSyncDTO.getEndDate()+"】");
                    //查询该表的全部分区字段
                    hdfsList = hiveToOracleService.selectHivePkeyName(tblName,tblDatabaseName);
                    //循环匹配分区字段配置表 拿到字段名称和字段格式
                    if(ObjectUtil.isNotEmpty(hdfsList)&&hdfsList.size()>0){
                        for (String h : hdfsList) {
                            LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                            timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                            timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, h).eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                            TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
                            //不为空就匹配成功
                            if (ObjectUtils.isNotEmpty(matching)) {
                                //获取当前时间分区内的hdfs地址
                                locationList = hiveToOracleService
                                        .selectHiveTableHDFSBee(tblName, "", matching.getTimeFieldKay(), String.valueOf(handMoveSyncDTO.getStaDate()),String.valueOf(handMoveSyncDTO.getEndDate()), matching.getTimeFieldValue(), tblDatabaseName);
                                if(locationList!=null&&locationList.size()>0){
                                    locationList.forEach(lo -> {
                                        HdfsHeatBackup kup = new HdfsHeatBackup();
                                        //切割分区信息
                                        String partName = "";
                                        if (StrUtil.split(lo,tblName).length == 2) {
                                            partName = StrUtil.split(lo, tblName)[1];
                                        } else {
                                            log.warn("===========" + lo + "===========");
                                        }
                                        lo = wGCluster.getHdfsPrefixPath() + tblDatabaseName + ".db/" + tblName + partName;
                                        kup.setSourceAddress(lo);
                                        kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, wGCluster.getHdfsPrefixPath(), uatCluster.getHdfsPrefixPath()), "/" + tblDatabaseName + ".db/", "/" + prefix + tblDatabaseName + ".db/"));
                                        kup.setSourceDbName(tblDatabaseName);
                                        kup.setTargetDbName(prefix + tblDatabaseName);
                                        kup.setSourceTableName(tblName);
                                        kup.setTargetTableName(tblName);
                                        hdfsHeatBackups.add(kup);
                                    });
                                }else{
                                    map.put("error","当前时间区间内无HDFS信息");
                                }
                                break;
                            }else{
                                //查询该表的hdfs路径
                                List<String> hdfsAlsit = hiveToOracleService.selectHiveLocationUat(tblDatabaseName,tblName);
                                if(ObjectUtil.isNotEmpty(hdfsAlsit)){
                                    for (String lo: hdfsAlsit) {
                                        getHeatBackup(handMoveSyncDTO,hdfsHeatBackups,lo,wGCluster,uatCluster,tblDatabaseName,prefix,tblName);
                                    }
                                }
                            }
                        }
                    }else{
                        //查询该表的hdfs路径
                        List<String> hdfsAlsit = hiveToOracleService.selectHiveLocationUat(tblDatabaseName,tblName);
                        if(ObjectUtil.isNotEmpty(hdfsAlsit)){
                            for (String lo: hdfsAlsit) {
                                getHeatBackup(handMoveSyncDTO,hdfsHeatBackups,lo,wGCluster,uatCluster,tblDatabaseName,prefix,tblName);
                            }
                        }
                    }
                }
            }else{
                mess.setMessageSubject(ScanCommonConstants.ClusterName.CSSJTB.value+"~"+StringUtils.toUpperCase(handMoveSyncDTO.getDbName()));
                messageCenterService.saveOrUpdate(mess);
                //库级同步 查询表信息
                uatSyncLogsTask.setTaskName(StringUtils.toUpperCase(handMoveSyncDTO.getDbName())+DateUtil.format(new Date(), "yyyyMMdd"));
                uatSyncLogsTask.setSourceName("【"+handMoveSyncDTO.getLevel()+"】【增量】【"+handMoveSyncDTO.getStaDate()+"--"+handMoveSyncDTO.getEndDate()+"】");
                //黑名单过滤
                if(ObjectUtil.isNotEmpty(handMoveSyncDTO.getUatBlackListList())&&handMoveSyncDTO.getUatBlackListList().size()>0){
                    //拼接条件
                    where = this.whereLike(handMoveSyncDTO.getUatBlackListList());
                }
                scanMetadata = hiveToOracleService.selectHiveTableList(handMoveSyncDTO.getDbName(),where);
                if(ObjectUtil.isNotEmpty(scanMetadata)&&scanMetadata.size()>0){
                    for (ScanMetadata s: scanMetadata) {
                        //查询该表的全部分区字段
                        hdfsList = hiveToOracleService.selectHivePkeyName(s.getTblName(),handMoveSyncDTO.getDbName());
                        //循环匹配分区字段配置表 拿到字段名称和字段格式
                        if(ObjectUtil.isNotEmpty(hdfsList)){
                            log.warn("===================有分区查询hdfs路径");
                            for (String h : hdfsList) {
                                LambdaQueryWrapper<TimeFieldMatching> timeFieldWrapper = Wrappers.lambdaQuery();
                                timeFieldWrapper.select(TimeFieldMatching::getTimeFieldKay, TimeFieldMatching::getTimeFieldValue);
                                timeFieldWrapper.eq(TimeFieldMatching::getTimeFieldKay, h).eq(TimeFieldMatching::getDel, CommonConstants.STATUS_NORMAL);
                                TimeFieldMatching matching = timeFieldMatchingService.getOne(timeFieldWrapper);
                                //不为空就匹配成功
                                if (ObjectUtils.isNotEmpty(matching)) {
                                    //获取当前时间分区内的hdfs地址
                                    locationList = hiveToOracleService.selectHiveTableHDFSBee(s.getTblName(), "", matching.getTimeFieldKay(), String.valueOf(handMoveSyncDTO.getStaDate()),String.valueOf(handMoveSyncDTO.getEndDate()), matching.getTimeFieldValue(), handMoveSyncDTO.getDbName());
                                    if(locationList!=null&&locationList.size()>0){
                                        locationList.forEach(lo -> {
                                            HdfsHeatBackup kup = new HdfsHeatBackup();
                                            //切割分区信息
                                            String partName = "";
                                            if (StrUtil.split(lo, s.getTblName()).length == 2) {
                                                partName = StrUtil.split(lo, s.getTblName())[1];
                                            } else {
                                                log.warn("===========" + lo + "===========");
                                            }
                                            lo = wGCluster.getHdfsPrefixPath() + handMoveSyncDTO.getDbName() + ".db/" + s.getTblName() + partName;
                                            kup.setSourceAddress(lo);
                                            kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, wGCluster.getHdfsPrefixPath(), uatCluster.getHdfsPrefixPath()), "/" + handMoveSyncDTO.getDbName() + ".db/", "/" + prefix + handMoveSyncDTO.getDbName() + ".db/"));
                                            kup.setSourceDbName(handMoveSyncDTO.getDbName());
                                            kup.setTargetDbName(prefix + handMoveSyncDTO.getDbName());
                                            kup.setSourceTableName(s.getTblName());
                                            kup.setTargetTableName(s.getTblName());
                                            hdfsHeatBackups.add(kup);
                                        });
                                    }else{
                                        map.put("error","当前时间区间内无HDFS信息");
                                    }
                                    break;
                                }else{
                                    //查询该表的hdfs路径
                                    List<String> hdfsAlsit = hiveToOracleService.selectHiveLocationUat(handMoveSyncDTO.getDbName(),s.getTblName());
                                    if(ObjectUtil.isNotEmpty(hdfsAlsit)){
                                        for (String lo: hdfsAlsit) {
                                            getHeatBackupDbName(handMoveSyncDTO,hdfsHeatBackups,s,lo,wGCluster,uatCluster,prefix);
                                        }
                                    }
                                }
                            }
                        }else{
                            log.warn("===================无分区查询hdfs路径");
                            //查询该表的hdfs路径
                            List<String> hdfsAlsit = hiveToOracleService.selectHiveLocationUat(handMoveSyncDTO.getDbName(),s.getTblName());
                            if(ObjectUtil.isNotEmpty(hdfsAlsit)){
                                for (String lo: hdfsAlsit) {
                                    log.warn("===================【"+handMoveSyncDTO.getDbName()+"库】"+s.getTblName()+"==hdfs路径"+lo);
                                    getHeatBackupDbName(handMoveSyncDTO,hdfsHeatBackups,s,lo,wGCluster,uatCluster,prefix);
                                }
                            }
                        }
                    }
                }else{
                    uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                    uatSyncLogsTaskService.saveOrUpdate(uatSyncLogsTask);
                    UatSyncLogs logs = new UatSyncLogs();
                    logs.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                    logs.setTaskName(uatSyncLogsTask.getTaskName());
                    logs.setTaskId(uatSyncLogsTask.getId());
                    logs.setLog("\n["+DateUtil.now()+"]【"+handMoveSyncDTO.getDbName()+"库下暂无表信息】或者【"+handMoveSyncDTO.getDbName()+"库不存在】");
                    uatSyncLogsService.saveOrUpdate(logs);
                    map.put("error","当前库下暂无表信息");
                    return map;
                }
            }
        }else{
            map.put("error","库名不能为空");
            return map;
        }
        //将比对sql对象 赋值给传输DTO
        ruleToDistCPDTO.setMessageId(mess.getId());
        ruleToDistCPDTO.setHdfsHeatBackups(new ArrayList<>(hdfsHeatBackups));
        Set<String> set = new HashSet<>();
        for (HdfsHeatBackup hdfs: hdfsHeatBackups) {
            set.add(hdfs.getSourceName());
        }
        for (String name : set) {
            if(StrUtil.isNotBlank(uatSyncLogsTask.getLibraryTableName())){
                uatSyncLogsTask.setLibraryTableName(uatSyncLogsTask.getLibraryTableName()+"\n"+name);
            }else{
                uatSyncLogsTask.setLibraryTableName(name);
            }
        }
        uatSyncLogsTask.setTableHdfsNumber(""+hdfsHeatBackups.size());
        uatSyncLogsTaskService.saveOrUpdate(uatSyncLogsTask);
        ruleToDistCPDTO.setUatSyncLogsTask(uatSyncLogsTask);
        String s  = "";
        //不为空且的等于“是”
        if(StrUtil.isNotBlank(handMoveSyncDTO.getCpWhether())&&"是".equals(handMoveSyncDTO.getCpWhether())){
            UatSyncLogs uatSyncLogs = new UatSyncLogs();
            if(ObjectUtil.isNotEmpty(ruleToDistCPDTO.getHdfsHeatBackups())&&ruleToDistCPDTO.getHdfsHeatBackups().size() > 0) {
                s = ruleToDistCPService.heatBackupWgToUat(ruleToDistCPDTO);
            }else{
                //当前结束时间
                Date endDate = DateUtil.date();
                uatSyncLogs.setTaskId(uatSyncLogsTask.getId());
                uatSyncLogs.setStandbyTime(DateUtil.formatBetween(staDate,endDate, BetweenFormater.Level.SECOND));
                uatSyncLogs.setLog("\n["+DateUtil.now()+"]【HDFS路径集合为空】");
                uatSyncLogs.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                uatSyncLogsService.saveOrUpdate(uatSyncLogs);
                log.warn("===================HDFS路径集合为空");
            }
        }
        //当前结束时间
        Date endDate = DateUtil.date();
        uatSyncLogsTask.setStandbyTime(DateUtil.formatBetween(staDate,endDate, BetweenFormater.Level.SECOND));
        if(StrUtil.isNotBlank(s)&&"成功".equals(s)){
            map.put("success","同步完成");
            uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            mess.setDataType(ScanCommonConstants.ClusterName.YWC.value);
        }else{
            map.put("error","同步异常");
            uatSyncLogsTask.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            mess.setDataType(ScanCommonConstants.ClusterName.YC.value);
        }
        uatSyncLogsTaskService.updateById(uatSyncLogsTask);
        messageCenterService.updateById(mess);
        return map;
    }


    /**
     * 黑名单
     * @param whereBlackList
     * @return
     */
    public String whereLike(List<String> whereBlackList){

        String whereLikeBlackTOP = "and not (";
        String whereLikeLO = " )";
        String whereLikeBlack = "";
        //黑名单
        if(whereBlackList!=null&&whereBlackList.size()>0){
            for (String wh: whereBlackList) {
                if(StrUtil.isBlank(whereLikeBlack)){
                    whereLikeBlack = "t.TBL_NAME like '" + wh + "' escape '\\' ";
                }else{
                    whereLikeBlack += " or t.TBL_NAME like '" + wh + "' escape '\\' ";
                }
            }
            whereLikeBlack = whereLikeBlackTOP + whereLikeBlack + whereLikeLO;
        }
        return whereLikeBlack;
    }

    /**
     * 获取热备所需HDFS路径（表级）
     * @param handMoveSyncDTO
     * @param hdfsHeatBackups
     * @param lo
     * @param wGCluster
     * @param uatCluster
     * @param tblDatabaseName
     * @param prefix
     * @param tblName
     */
    public void getHeatBackup(HandMoveSyncDTO handMoveSyncDTO,Set<HdfsHeatBackup> hdfsHeatBackups,String lo,ClusterSparingConf wGCluster,ClusterSparingConf uatCluster,String tblDatabaseName,String prefix,String tblName){
        HdfsHeatBackup kup = new HdfsHeatBackup();
        String  aresoftLo = StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),wGCluster.getHdfsPrefixPath().substring(0,getSlashIndex(wGCluster.getHdfsPrefixPath(),3)));
        kup.setSourceAddress(aresoftLo);
        kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),uatCluster.getHdfsPrefixPath().substring(0,getSlashIndex(uatCluster.getHdfsPrefixPath(),3))), "/" + tblDatabaseName , "/" + prefix + tblDatabaseName ));
        kup.setSourceDbName(tblDatabaseName);
        kup.setTargetDbName(prefix + tblDatabaseName);
        kup.setSourceTableName(tblName);
        kup.setTargetTableName(tblName);
        hdfsHeatBackups.add(kup);
    }


    /**
     * 获取热备所需HDFS路径（库级）
     * @param handMoveSyncDTO
     * @param hdfsHeatBackups
     * @param s
     * @param lo
     * @param wGCluster
     * @param uatCluster
     * @param prefix
     */
    public void getHeatBackupDbName(HandMoveSyncDTO handMoveSyncDTO,Set<HdfsHeatBackup> hdfsHeatBackups,ScanMetadata s,String lo,ClusterSparingConf wGCluster,ClusterSparingConf uatCluster,String prefix){
        HdfsHeatBackup kup = new HdfsHeatBackup();
        String  aresoftLo = StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),wGCluster.getHdfsPrefixPath().substring(0,getSlashIndex(wGCluster.getHdfsPrefixPath(),3)));
        kup.setSourceAddress(aresoftLo);
        kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo,lo.substring(0,getSlashIndex(lo,3)),uatCluster.getHdfsPrefixPath().substring(0,getSlashIndex(uatCluster.getHdfsPrefixPath(),3))), "/" + handMoveSyncDTO.getDbName() , "/" + prefix + handMoveSyncDTO.getDbName() ));
        kup.setSourceDbName(handMoveSyncDTO.getDbName());
        kup.setTargetDbName(prefix + handMoveSyncDTO.getDbName());
        kup.setSourceTableName(s.getTblName());
        kup.setTargetTableName(s.getTblName());
        hdfsHeatBackups.add(kup);
    }

    /**
     * 获取第几个斜杆所在位置
     * @return
     */
    public int getSlashIndex(String url,int num){
        int slashCount = 0; // 记录斜杆的数量
        int slashIndex = -1; // 记录第几个斜杆的位置
        for (int i = 0; i < url.length(); i++) {
            if (url.charAt(i) == '/') {
                slashCount++;
                if (slashCount == num) {
                    slashIndex = i;
                    break;
                }
            }
        }
        return slashIndex;
    }
}
